package cn.codenest.util.impl;

import cn.codenest.listener.MyConfirmListener;
import cn.codenest.listener.MyReturnListener;
import cn.codenest.util.RbmqService;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;

/**
 * @author ：Hyman
 * @date ：Created in 2021/7/6 12:10
 * @description：
 * @modified By：
 * @version: $
 */
public class RbmqServiceImpl implements RbmqService {

    private String rabbitMqHost = "127.0.0.1";
    private String rabbitMqUsername = "guest";
    private String rabbitMqPassword = "guest";
    private Integer rabbitMqPort = 5672;
    private Integer rabbitMqTimeOut = 10000;


    @Override
    public void sendMsgEachBlockingConform(String msg) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(this.rabbitMqHost);
        factory.setPort(this.rabbitMqPort);
        factory.setConnectionTimeout(this.rabbitMqTimeOut);
        factory.setUsername(this.rabbitMqUsername);
        factory.setPassword(this.rabbitMqPassword);
        factory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            try {
                connection = factory.newConnection();
                channel = connection.createChannel();
                channel.exchangeDeclare("direct-exchange", "direct", true, false, null);
                channel.queueDeclare("test-queue", true, false, false, null);
                com.rabbitmq.client.AMQP.Queue.BindOk ok = channel.queueBind("test-queue", "direct-exchange", "test-queue");
                /*开启事务，事务可以保证中间所有的消息全部发送成功或者全部回滚，但是性能很低*/
                //channel.txSelect();
                channel.confirmSelect();
                for (int i = 0; i < 10; i++) {
                    String newMsg = msg + i;
                    channel.basicPublish("direct-exchange", "test-queue", null, newMsg.getBytes("UTF-8"));
                    if (channel.waitForConfirms()) {
                        System.out.println("发送成功");
                    } else {
                        System.out.println("消息发送失败");
                        saveToDB(newMsg);
                    }
                }
                /*提交事务*/
                //channel.txCommit();
            } catch (IOException e) {
                System.out.println("服务器拒绝连接。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (TimeoutException e) {
                System.out.println("连接服务器超时。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (ShutdownSignalException e) {
                System.out.println("交换机故障或者不存在。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } finally {
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            closeChanelAndConnection(connection, channel);
        }

    }

    @Override
    public void sendMsgMultiBlockingConform(String msg) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(this.rabbitMqHost);
        factory.setPort(this.rabbitMqPort);
        factory.setConnectionTimeout(this.rabbitMqTimeOut);
        factory.setUsername(this.rabbitMqUsername);
        factory.setPassword(this.rabbitMqPassword);
        factory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            try {
                connection = factory.newConnection();
                channel = connection.createChannel();
                channel.exchangeDeclare("direct-exchange", "direct", true, false, null);
                channel.queueDeclare("test-queue", true, false, false, null);
                com.rabbitmq.client.AMQP.Queue.BindOk ok = channel.queueBind("test-queue", "direct-exchange", "test-queue");
                channel.confirmSelect();
                for (int i = 0; i < 10; i++) {
                    String newMsg = msg + i;
                    channel.basicPublish("direct-exchange", "test-queue", null, newMsg.getBytes("UTF-8"));
                }
                channel.waitForConfirmsOrDie();
            } catch (IOException e) {
                System.out.println("服务器拒绝连接。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (TimeoutException e) {
                System.out.println("连接服务器超时。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (ShutdownSignalException e) {
                System.out.println("交换机故障或者不存在。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } finally {
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            closeChanelAndConnection(connection, channel);
        }
    }

    @Override
    public void sendMsgAsyConform(String msg) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(this.rabbitMqHost);
        factory.setPort(this.rabbitMqPort);
        factory.setConnectionTimeout(this.rabbitMqTimeOut);
        factory.setUsername(this.rabbitMqUsername);
        factory.setPassword(this.rabbitMqPassword);
        factory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            try {
                connection = factory.newConnection();
                channel = connection.createChannel();
                channel.exchangeDeclare("direct-exchange", "direct", true, false, null);
                channel.queueDeclare("test-queue", true, false, false, null);
                com.rabbitmq.client.AMQP.Queue.BindOk ok = channel.queueBind("test-queue", "direct-exchange", "test-queue");
                channel.confirmSelect();
                HashMap<Long, String> map = new HashMap(32);
                for (int i = 0; i < 10; i++) {
                    String newMsg = msg + i;
                    channel.basicPublish("direct-exchange", "test-queue", null, newMsg.getBytes("UTF-8"));
                    map.put((long) (i + 1), newMsg);
                }
                channel.addConfirmListener(new MyConfirmListener(map));
            } catch (IOException e) {
                System.out.println("服务器拒绝连接。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (TimeoutException e) {
                System.out.println("连接服务器超时。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (ShutdownSignalException e) {
                System.out.println("交换机故障或者不存在。");
                e.printStackTrace();
                saveToDB(msg);
            }
        } finally {
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            closeChanelAndConnection(connection, channel);
        }
    }

    @Override
    public void sendMsgHandlingReturn(String msg) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(this.rabbitMqHost);
        factory.setPort(this.rabbitMqPort);
        factory.setConnectionTimeout(this.rabbitMqTimeOut);
        factory.setUsername(this.rabbitMqUsername);
        factory.setPassword(this.rabbitMqPassword);
        factory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            try {
                connection = factory.newConnection();
                channel = connection.createChannel();
                channel.exchangeDeclare("direct-exchange", "direct", true, false, null);
                channel.queueDeclare("test-queue", true, false, false, null);
                com.rabbitmq.client.AMQP.Queue.BindOk ok = channel.queueBind("test-queue", "direct-exchange", "test-queue");
                channel.basicPublish("direct-exchange", "test-queue", true, false, MessageProperties.PERSISTENT_BASIC, msg.getBytes("UTF-8"));
                channel.addReturnListener(new MyReturnListener());
            } catch (IOException e) {
                System.out.println("服务器拒绝连接。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (TimeoutException e) {
                System.out.println("连接服务器超时。");
                e.printStackTrace();
                saveToDB(msg);
            } catch (ShutdownSignalException e) {
                System.out.println("交换机故障或者不存在。");
                e.printStackTrace();
                saveToDB(msg);
            }
        } finally {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            closeChanelAndConnection(connection, channel);
        }
    }

    private void saveToDB(String msg) {
        System.out.println("消息保存到数据库成功:" + msg);
    }

    private void closeChanelAndConnection(Connection connection, Channel channel) {
        if (channel != null && channel.isOpen()) {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
        if (connection != null && connection.isOpen()) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
